package com.linran.webjsonrsp.controller;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.Map;
import java.util.concurrent.*;

@Slf4j
@RequestMapping("/poll")
@RestController
public class LongPollController {
    final String taskId = "taskId";
    public static final ThreadPoolExecutor executor = new ThreadPoolExecutor(3,
            3,
            3,
            TimeUnit.MINUTES,
            new LinkedBlockingQueue<>(),
            new ThreadPoolExecutor.AbortPolicy());

    public static volatile Map<String, Future<TaskVo>> map = new ConcurrentHashMap<>();

    /**
     * 当前框架第一个线程进来后会阻塞其他线程一段时间，有点问题，建议把当前类拷贝到其他框架测试
     */
    @GetMapping("/info")
    public void info() {
//        final String taskId = "taskId"+System.currentTimeMillis();
        log.info(String.format("%s befor put map contains key %s  %s  %n", Thread.currentThread().getId(), map.containsKey(taskId), JSON.toJSONString(map.keySet())));
        TaskVo taskVo = null;

//        DeferredResult<TaskVo> deferredResult = new DeferredResult<>(1000 * 30L,//超时时间
//                new Object()); //超时相应对象
//        deferredResult.onCompletion(() -> {
//            //请求完成处理事情
//            System.out.println("do something on completion.");
//            map.remove(taskId);
//        });

        /**主逻辑*/
        try {
            Future<TaskVo> future;
//            synchronized (LongPollController.class) {
            if (!map.containsKey(taskId)) {
                // 新建
                CustomerCallable customerCallable = new CustomerCallable("lin", "ran");
                future = executor.submit(customerCallable);
                // 维护任务
                map.put(taskId, future);
                log.info(String.format("after put map contains key %s %s  %n", map.containsKey(taskId), JSON.toJSONString(map.keySet())));
            } else {
                // 拿到历史任务
                future = map.get(taskId);
            }
//            }
            // 拿到结果
            log.info("开始阻塞.");
            taskVo = future.get();
            taskVo.setMessage(taskVo.getMessage() + taskId);
            log.info(String.format("hello word.%s  %s  %n", JSON.toJSONString(taskVo), future));
        } catch (RejectedExecutionException rejectedException) {
            rejectedException.printStackTrace();
            System.out.println("任务拒绝添加");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("其他异常");
        }

        // 调用此方法时立即向浏览器发出响应；未调用时请求被挂起
//        deferredResult.setResult(taskVo);
//        return deferredResult;
        map.remove(taskId);
    }
}

class CustomerCallable implements Callable<TaskVo> {

    private String operA;
    private String operB;

    public CustomerCallable(String operA, String operB) {
        this.operA = operA;
        this.operB = operB;
    }

    @Override
    public TaskVo call() throws Exception {
        System.out.println("call the thread.****************开始任务");
        TimeUnit.SECONDS.sleep(30);
        TaskVo taskVo = new TaskVo();
        taskVo.setCode(200);
        taskVo.setMessage(operA + operB);
        System.out.println("call the thread.****************结束任务");
        return taskVo;
    }
}

@Setter
@Getter
class TaskVo {
    private int code;
    private String message;
}
